/**
  * @author marinapopova
  * Feb 24, 2016
 */
package com.yd.spark.demo;

import java.io.BufferedReader;
import java.io.IOException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Map;

import org.apache.commons.io.Charsets;
import org.apache.commons.io.IOUtils;

import com.google.common.collect.Maps;

/**
 * 
 * This is an example of a customized Message Handler - by implementing IMessageHandler interface
 * and using the ElasticSearchBatchService to delegate most of the non-customized logic
 *
 */
public class SimpleMessageHandlerImpl{
	
	public static final String[] fieldNames = new String[]{"datetime","null1","service_order_id","round","batch","flag","driver_id","distance","dispatch_time","dispatch_lat","dispatch_lng","dispatch_total_rate","dispatch_snapshot","response_time","accept_status","response_lat","response_lng","response_distance","response_time_length","decision_time","decision_total_rate","decision_result","decision_failure_reason","decision_msg_snapshot","subtract_amount","add_price_set","response_snapshot","is_assigned","route_distance","route_time_length","distance_time_length"};
	public static String parseToJson(byte[] inputMessage) throws IOException{
		String line = UnicodeUtils.convertUnicode(inputMessage);
		Map<String , String> jsonBuild = Maps.newHashMap();
		
		//log分割成13份, 取出前12个字段
		String[] fields = line.split(" ",13);
		for(int i=0; i< 12; i++){
			jsonBuild.put(fieldNames[i],fields[i]);
		}
		//继续将第13份分割成2份,取出 dispatch_snapshot(第13个)字段
		fields = fields[12].split("} ", 2);
		jsonBuild.put(fieldNames[12],fields[0]+"}");
		//继续分割剩下的字段
		fields = fields[1].split(" ", -1);
		for(int i=0; i< fields.length; i++){
			jsonBuild.put(fieldNames[i+13],fields[i]);
		}
		System.out.println(jsonBuild.size());
		return JsonUtils.toJson(jsonBuild);
	}
	
	public static void main(String[] args) throws IOException {
		
		Path path = Paths.get("dispatch_detail_info.txt");
		BufferedReader in = java.nio.file.Files.newBufferedReader(path, Charsets.UTF_8);
		String line = null;
		while ((line = in.readLine()) != null) {
			String[] fields = line.split(" ", -1);
			if(fields.length != 31){
				System.err.println(UnicodeUtils.convertUnicode(line));
				parseToJson(line.getBytes());
			}
			
		}
		IOUtils.closeQuietly(in);		
	}
}
